package com.atguigu.day04;

import com.atguigu.utils.IntegerSource;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Comparator;

public class Example2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new IntegerSource())
                .keyBy(r -> r % 2)
                .process(new HistoryDataSort())
                .print();

        env.execute();
    }

    public static class HistoryDataSort extends KeyedProcessFunction<Integer, Integer, String> {
        private ListState<Integer> history;

        @Override
        public void open(Configuration parameters) throws Exception {
            history = getRuntimeContext().getListState(
                    new ListStateDescriptor<Integer>(
                            "history",
                            Types.INT
                    )
            );
        }

        @Override
        public void processElement(Integer in, Context ctx, Collector<String> out) throws Exception {
            // 添加到输入数据的key对应的列表中
            // 例如输入数据是3，那么in被添加到key为1对应的列表中
            history.add(in);

            String result = "";
            if (ctx.getCurrentKey() == 0) {
                result += "偶数历史数据：";
            } else {
                result += "奇数历史数据：";
            }

            ArrayList<Integer> integers = new ArrayList<>();
            for (Integer i : history.get()) integers.add(i);

            integers.sort(new Comparator<Integer>() {
                @Override
                public int compare(Integer i1, Integer i2) {
                    return i1 - i2;
                }
            });

            for (Integer i : integers) {
                result += i + " => ";
            }

            out.collect(result);
        }
    }
}
